package apps.sim.util.thread;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import apps.sim.util.thread.TaskResponse.ExecutionStatus;

public class GenericTaskExecutor {
	private static final int DEFAULT_POOL_SHUTDOWN_TIME = 20;

	private ThreadPoolExecutor executorService = null;
	private LinkedBlockingQueue<Runnable> taskQueue = null;

	public ThreadPoolExecutor getExecutorService() {
		return executorService;
	}

	public GenericTaskExecutor(String executorServiceName) {
		this(executorServiceName, 50, 50);
	}

	public GenericTaskExecutor(String executorServiceName, int coreThreadCount, int maxPoolSize) {
		executorService = getConfiguredExecutor(executorServiceName, coreThreadCount, maxPoolSize);
		createShutdownHook();
	}

	private void createShutdownHook() {
		Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
			public void run() {
				if (executorService != null && !executorService.isShutdown() && !executorService.isTerminated()) {
					executorService.shutdown();
					try {
						executorService.awaitTermination(DEFAULT_POOL_SHUTDOWN_TIME, TimeUnit.SECONDS);
					} catch (Throwable t) {
						// eat it
					}
				}

			}
		}));
	}

	private ThreadPoolExecutor getConfiguredExecutor(String executorServiceName, int coreThreadCount, int maxPoolSize) {
		int keepAliveSeconds = 10;
		taskQueue = new LinkedBlockingQueue<Runnable>();
		final ThreadFactory threadFactory = new AThreadFactory(executorServiceName);
		final ThreadPoolExecutor g = new ThreadPoolExecutor(coreThreadCount, maxPoolSize, keepAliveSeconds, TimeUnit.SECONDS, taskQueue, threadFactory);
		return g;
	}

	public <T> List<Future<T>> execute(TaskContainer<T> container, long timeoutSeconds, CountDownLatch latch, boolean cancelAfterTimeOut, boolean async) {
		final List<Future<T>> executeResults = new ArrayList<Future<T>>();
		final List<CallableTask<T>> tasks = container.tasks();
		for (CallableTask<T> task : tasks) {
			task.setLatch(latch);
			Future<T> future = executorService.submit(task);
			executeResults.add(future);
		}
		return executeResults;
	}

	public <T> List<TaskResponse<T>> execute(TaskContainer<T> container, long timeoutSeconds, boolean cancelAfterTimeOut, boolean async) {
		final CountDownLatch latch = new CountDownLatch(container.taskCount());
		final List<Future<T>> executeResults = new ArrayList<Future<T>>();
		final List<CallableTask<T>> tasks = container.tasks();
		final ArrayList<T> retryResults = new ArrayList<T>();
		for (CallableTask<T> task : tasks) {
			task.setLatch(latch);
			Future<T> future = executorService.submit(task);
			executeResults.add(future);
		}

		cancelRunningTasks(executeResults, latch, timeoutSeconds);
		final List<TaskResponse<T>> results = new ArrayList<TaskResponse<T>>();
		for (int index = 0; index < executeResults.size(); index++) {
			final Future<T> jobResult = executeResults.get(index);
			T response = null;
			Exception exception = null;
			ExecutionStatus status = ExecutionStatus.SUCCESS;
			try {
				response = jobResult.get();
			} catch (CancellationException e) {
				status = ExecutionStatus.CANCELED;
				exception = e;
			} catch (InterruptedException e) {
				e.printStackTrace();
				status = ExecutionStatus.INTERRUPTED;
				exception = e;
			} catch (ExecutionException e) {
				e.printStackTrace();
				status = ExecutionStatus.ERROR;
				exception = e;
			}

			final TaskResponse<T> jobResponse = new TaskResponse<T>();
			jobResponse.setExecutionStatus(status);
			jobResponse.setResponse(response);
			if (response == null && retryResults.size() > index && retryResults.get(index) != null) {
				jobResponse.setResponse(retryResults.get(index));
			}
			jobResponse.setException(exception);
			results.add(jobResponse);
		}

		return results;
	}

	public <T> void fireAndForget(TaskContainer<T> container) {
		final CountDownLatch latch = new CountDownLatch(0);
		final List<CallableTask<T>> tasks = container.tasks();
		for (CallableTask<T> task : tasks) {
			task.setLatch(latch);
			executorService.submit(task);
		}
		try {
			latch.await(1, TimeUnit.MILLISECONDS);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	private <T> void cancelRunningTasks(List<Future<T>> executeResults, CountDownLatch doneSignal, long timeoutSeconds) {
		boolean finished = false;
		try {
			finished = doneSignal.await(timeoutSeconds, TimeUnit.SECONDS);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

		if (!finished) {
			for (Future<T> future : executeResults) {
				if (!future.isDone()) {
					future.cancel(true);
				}
			}
		}
	}

	private static class AThreadFactory implements ThreadFactory {
		final ThreadGroup group;
		final AtomicInteger threadNumber = new AtomicInteger(1);
		final String namePrefix;

		AThreadFactory(String name) {
			SecurityManager s = System.getSecurityManager();
			group = s != null ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
			namePrefix = "Pool-" + name + "-thread-";
		}

		public Thread newThread(Runnable r) {
			Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
			if (t.isDaemon()) {
				t.setDaemon(false);
			}
			if (t.getPriority() != Thread.NORM_PRIORITY) {
				t.setPriority(Thread.NORM_PRIORITY);
			}
			return t;
		}
	}

}
